package cn.doitedu.rtdw.customer.topn;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/12/18
 * @Desc: 学大数据，上多易教育
 *   最近60分钟，各品类浏览量最大的前10个商品
 **/
public class Job01_CategoryTopnHotViewProduct {
    public static void main(String[] args) {
        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 创建kafka中dwd行为明细topic的映射表
        tenv.executeSql(
                " create table dwd_events_kafka (                                 "+
                        "     event_id     string                                 "+
                        "     ,action_time   bigint                               "+
                        "     ,properties  map<string,string>                     "+
                        "     ,page_type string                                   "+
                        "     ,rt as to_timestamp_ltz(action_time,3)              "+
                        "     ,watermark for rt as rt - interval '0' second       "+
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    " +
                        "     'properties.group.id' = 'doit43-2',                 " +
                        "     'scan.startup.mode' = 'latest-offset',              " +
                        "     'value.format'='json',                              " +
                        "     'value.json.fail-on-missing-field'='false',         " +
                        "     'value.fields-include' = 'EXCEPT_KEY'   )           "
        );


        // 创建hbase中的商品信息维表的映射表
        tenv.executeSql(
                         " create table dim_product_info_hbase(             "+
                        "     item_id string,                               "+
                        " 	  f  row<item_name string,category_id string>   "+
                        " )  WITH (                                         "+
                        "    'connector' = 'hbase-2.2',                     "+
                        "    'table-name' = 'dim_product_info',             "+
                        "    'zookeeper.quorum' = 'doitedu:2181'            "+
                        " )                                                 "
        );



        // 统计sql
        tenv.executeSql(
                " with a AS (                                                                                               "+
                        "     SELECT                                                                                                "+
                        "         event_id,                                                                                         "+
                        "         action_time,                                                                                      "+
                        "         page_type,                                                                                        "+
                        "         properties['item_id'] as item_id,                                                "+
                        " 		rt,                                                                                                 "+
                        " 		proctime() as pt                                                                                    "+
                        "     FROM dwd_events_kafka                                                                                 "+
                        "     WHERE event_id = 'page_load' AND page_type='商品详情页'                                               "+
                        " )                                                                                                         "+
                        " , tmp AS (                                                                                                "+
                        "     SELECT                                                                                                "+
                        "        b.category_id,                                                                                     "+
                        "        a.item_id,                                                                                         "+
                        " 	   a.rt                                                                                                 "+
                        "     FROM a                                                                                                "+
                        "     LEFT JOIN dim_product_info_hbase FOR SYSTEM_TIME AS OF a.pt   as b                                    "+
                        "     ON a.item_id = b.item_id                                                                              "+
                        " )                                                                                                         "+
                        " ,tmp2 AS (                                                                                                "+
                        "     SELECT                                                                                                "+
                        "         window_start,                                                                                     "+
                        "     	window_end,                                                                                         "+
                        "     	category_id,                                                                                        "+
                        "     	item_id,                                                                                            "+
                        " 		count(1) as view_cnt,                                                                               "+
                        "     	row_number() over(partition by window_start,window_end,category_id order by count(1) desc)  as rn   "+
                        "     from TABLE(                                                                                           "+
                        "         HOP(TABLE tmp,DESCRIPTOR(rt),INTERVAL '1' MINUTE,INTERVAL '1' HOUR)                               "+
                        "     )                                                                                                     "+
                        "     GROUP BY                                                                                              "+
                        "         window_start,                                                                                     "+
                        "     	window_end,                                                                                         "+
                        "     	category_id,                                                                                        "+
                        "     	item_id                                                                                             "+
                        " )	                                                                                                        "+
                        "                                                                                                           "+
                        " SELECT                                                                                                    "+
                        "   category_id,                                                                                            "+
                        " 	rn,                                                                                                     "+
                        " 	item_id,                                                                                                "+
                        " 	view_cnt                                                                                                "+
                        " FROM tmp2                                                                                                 "+
                        " WHERE rn<=10                                                                                              "
        ).print();


    }


}
